package com.heima.article.stream;


import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.common.constants.article.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.mess.app.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;
import org.springframework.stereotype.Component;

/**
 * @作者 itcast
 * @创建日期 2020/12/24 9:27
 **/
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {
    /**
     * 需要处理的数据来源 主题
     * @return
     */
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC;
    }
    /**
     * 处理完数据 发送到的 主题
     * @return
     */
    @Override
    public String sendTopic() {
        return HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC;
    }
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return stream.map((k,v)->{
                    // key:  ""   value: UpdateArticleMess对应的json
                    UpdateArticleMess updateArticleMess = JSON.parseObject(v, UpdateArticleMess.class);
                    log.info("流式处理==> 行为数据 ==> {}",updateArticleMess);
                    String key = updateArticleMess.getArticleId()+":"+ updateArticleMess.getType().name();
                    Integer value = updateArticleMess.getAdd();
                    // key:  123123143:VIEWS   value: 1
                    // key:  123123143:VIEWS   value: 1
                    // key:  123123143:VIEWS   value: 1
                    return new KeyValue<>(key,value.toString());
                })
                .groupByKey()
                .windowedBy(TimeWindows.of(10000))
                .count(Materialized.as("article_count"))
                .toStream()
                .map(
                        // key:  123123143:VIEWS   value: 3
                        (k,v)->new KeyValue<>(k.key(),formatObj(k.key(),v.toString()))
                );
    }
    // 封装消息 发送
    private String formatObj(String key, String value){
        // key:  123123143:VIEWS   value: 3
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        String[] split = key.split(":");
        mess.setArticleId(Long.valueOf(split[0]));
        if (split[1].equals(UpdateArticleMess.UpdateArticleType.LIKES.name())) {
            mess.setLike(Long.valueOf(value));
        }
        if (split[1].equals(UpdateArticleMess.UpdateArticleType.VIEWS.name())) {
            mess.setView(Long.valueOf(value));
        }
        if (split[1].equals(UpdateArticleMess.UpdateArticleType.COMMENT.name())) {
            mess.setComment(Long.valueOf(value));
        }
        if (split[1].equals(UpdateArticleMess.UpdateArticleType.COLLECTION.name())) {
            mess.setCollect(Long.valueOf(value));
        }
        log.info("流式处理后的数据==> {}",JSON.toJSONString(mess));
        return JSON.toJSONString(mess);
    }
}
